package com.azarias.dfs.namenode.server.component;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 这个组件，就是负责管理集群里的所有的datanode的
 *
 */
public class DataNodeManager {

	/**
	 * 内存中维护的datanode列表
	 */
	private ConcurrentHashMap<String,DataNodeInfo> datanodes = new ConcurrentHashMap<>();

	/**
	 * datanode进行注册
	 * @param ip 
	 * @param hostname
	 */
	public Boolean register(String ip, String hostname) {
		DataNodeInfo datanode = new DataNodeInfo(ip, hostname);
		String nodeName = ip+"-"+hostname;
		datanodes.put(nodeName,datanode);
		return true;
	}

	/**
	 * 心跳，用以移除通讯失败的组件
	 * @return
	 */
	public boolean heartbeat(String ip,String name){
		String nodeName = ip+"-"+name;
		DataNodeInfo dataNodeInfo = datanodes.get(nodeName);
		//如果没有被摘除就设置续约时间
		if( dataNodeInfo != null ){
			dataNodeInfo.setLatestHeartbeatTime(System.currentTimeMillis());
		}
		return true;
	}

	/**
	 * 后台处理datanode心跳下线
	 */
	class DataNodeAliveMonitor extends Thread {
		@Override
		public void run() {
			while (true){
				//创建一个需要剔除的node name数组
				List<String> needRemoveNameList = new ArrayList<>();
				try {
					//获取迭代器
					Iterator<DataNodeInfo> dataNodeInfoIterator = datanodes.values().iterator();
					while (dataNodeInfoIterator.hasNext()){
						 DataNodeInfo dataNodeInfo = dataNodeInfoIterator.next();
						 if(System.currentTimeMillis()-dataNodeInfo.getLatestHeartbeatTime()>90*1000){
						 	needRemoveNameList.add(dataNodeInfo.getIp()+"-"+dataNodeInfo.getHostname());
						 }

						 if(!needRemoveNameList.isEmpty()){
							 for (String name : needRemoveNameList) {
								 datanodes.remove(name);
							 }
						 }

					}

					Thread.sleep(1000);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
		}
	}
}
